package com.xxwy.zk.lock;

import com.xxwy.zk.entity.Lock;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.CreateMode;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @author wy
 * @date 2020/4/8 18:00
 */
public class ZookeeperLock {

  //链接zk服务器
  //创建临时序号节点
  private ZkClient zkClient;
  private String fPath;


  public ZookeeperLock(String fPath) {
    this.zkClient = new ZkClient("13.15.60.51:2181", 5000, 20000);
    this.fPath = fPath;
    zkClient.delete(fPath);
    String path = zkClient.create(fPath, "lock", CreateMode.PERSISTENT);
    System.out.printf("初始化path: %s \n", path);
  }

  private Lock createLockNode(String lockId) {
    String lockPath = fPath + "/" + lockId;
    String tPath = zkClient.createEphemeralSequential(lockPath, "w");
    Lock lock = new Lock(lockId, tPath);
    return lock;
  }

  //获得锁
  public Lock lock(String lockId, long timeout) {
    Lock lockNode = createLockNode(lockId);
    Lock lock = tryActiveLock(lockNode);
    if (!lock.isActive()) {
      synchronized (lock) {
        try {
          lockNode.wait(timeout);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }else {
      System.out.println(lock.getPath() + " 抢到锁了");
    }
    return lock;
  }

  //激活锁
  private Lock tryActiveLock(Lock lock) {
    List<String> children = zkClient.getChildren(fPath);
    List<String> filePaths = children.stream().sorted()
        .map(p -> fPath + "/" + p).collect(Collectors.toList());
    System.out.println(lock.getPath() + "尝试获取锁，现在抢锁队列" + filePaths);
    if (lock.getPath().equals(filePaths.get(0))) {
      lock.setActive(true);
    } else {
      //添加上一个节点的监听
      String beforePath = filePaths.get(filePaths.indexOf(lock.getPath()) - 1);
      zkClient.subscribeDataChanges(beforePath, new IZkDataListener() {
        @Override
        public void handleDataChange(String s, Object o) throws Exception {

        }

        @Override
        public void handleDataDeleted(String s) throws Exception {
          System.out.println(s + " 锁释放了");
          Lock lock1 = tryActiveLock(lock);
          synchronized (lock) {
            if (lock1.isActive()) {
              lock.notify();
            }
          }
          //取消监听节点的变化
          zkClient.unsubscribeDataChanges(beforePath, this);
        }
      });
    }
    return lock;
  }

  //释放锁
  public void unLock(Lock lock) {
    if (lock.isActive()) {
      zkClient.delete(lock.getPath());
      System.out.println(lock.getPath() + Thread.currentThread() + "bye");
    }
  }
}
